package com.intct.hbase;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Map;


public class windowcount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        FileSource<String> fs = FileSource.<String>forRecordStreamFormat(
                new TextLineFormat(),
                new Path("F:\\maxwell\\car.txt")
        ).build();
        DataStreamSource<String> ds = env.fromSource(
                fs,
                WatermarkStrategy.noWatermarks(),
                "myfile"
        );
        SingleOutputStreamOperator<Map<String, Object>> mapStream = ds.map(new MapFunction<String, Map<String, Object>>() {
            @Override
            public Map<String, Object> map(String line) throws Exception {
                Map<String, Object> map = JSON.parseObject(line);
                String timeStr = (String) map.get("time");
                LocalDateTime dateTime = LocalDateTime.parse(
                        timeStr,
                        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
                );
                long timestamp = dateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                map.put("timestamp", timestamp);
                return map;
            }
        });

        SingleOutputStreamOperator<Map<String, Object>> waterStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy
                .<Map<String, Object>>forBoundedOutOfOrderness(Duration.ofSeconds(300))
                .withTimestampAssigner((m, l) -> (long) m.get("timestamp")));

        SingleOutputStreamOperator<Tuple2<Map<String, Object>, Integer>> warnSpeed = waterStream.keyBy(map -> map.get("car"))
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .process(new ProcessWindowFunction<Map<String, Object>, Tuple2<Map<String, Object>, Integer>, Object, TimeWindow>() {

                    @Override
                    public void process(Object o, ProcessWindowFunction<Map<String, Object>, Tuple2<Map<String, Object>, Integer>, Object, TimeWindow>.Context context, Iterable<Map<String, Object>> iterable, Collector<Tuple2<Map<String, Object>, Integer>> collector) throws Exception {
                        int a = 0;
                        Map<String, Object> resultMap = null;

                        for (Map<String, Object> event : iterable) {
                            resultMap = event;
                            long speed = Long.parseLong(event.get("speed").toString());
                            if (speed > 60) {
                                a++;
                            }

                            Tuple2<Map<String, Object>, Integer> result = Tuple2.of(resultMap, a);

                            if (a >= 1) collector.collect(result);
                        }

                    }

                });


        SingleOutputStreamOperator<Tuple2<Map<String, Object>, Integer>> finalStream_over = warnSpeed.keyBy(t -> t.f0.get("car"))
                .reduce(new ReduceFunction<Tuple2<Map<String, Object>, Integer>>() {
                    @Override
                    public Tuple2<Map<String, Object>, Integer> reduce(Tuple2<Map<String, Object>, Integer> t1, Tuple2<Map<String, Object>, Integer> t2) throws Exception {
                        return Tuple2.of(t1.f0,t1.f1+t2.f1);
                    }

                });

        finalStream_over.print();
        env.execute();
        env.close();
    }
}



